package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 通过这个类来管理硬盘上的消息文件
 * */
public class MessageFileManager {

    // 定义一个内部类，用于表示消息队列的统计信息
    static public class Stat {
        public int totalCount;  // 消息总数
        public int validCount;  // 有效消息数
    }

    public void init() {
        // 暂时不需要额外的初始化工作
    }

    // 约定消息文件所在的目录和文件名
    // 获取指定队列对应的消息文件所在路径
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    // 获取指定队列对应的消息数据文件所在路径
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    // 获取指定队列对应的消息统计文件所在路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    // 每个队列都有自己的目录，每个目录下有两个文件：queue_data.txt和queue_stat.txt
    private Stat readStat(String queueName) {
        // 由于是文本文件，所以可以直接读取文件内容
        Stat stat = new Stat();
        try(InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();
            stat.validCount = scanner.nextInt();
            return stat;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void writeStat(String queueName, Stat stat) {
        // PrintWriter进行写文件
        // OutputStream打开文件，默认是覆盖写
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();
        }catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    // 创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        // 1. 队列对应的创建消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建队列目录失败! baseDir = " + baseDir.getAbsolutePath());
            }
        }
        // 2. 创建消息数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建队列数据文件失败! queueDataFile = " + queueDataFile.getAbsolutePath());
            }
        }
        // 3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建队列统计文件失败! queueStatFile = " + queueStatFile.getAbsolutePath());
            }
        }
        // 4. 初始化统计文件，0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    // 删除队列对应的文件和目录，队列也是可以被删除的，删除队列时，需要删除对应的文件和目录
    public void destroyQueueFiles(String queueName) throws IOException {
        // 1. 先删除文件，再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            throw new IOException("删除队列目录和文件失败! baseDir = " + baseDir.getAbsolutePath());
        }
    }

    // 检查队列的目录和文件是否存在
    //
    public boolean checkFileExits(String queueName) {
        // 判断队列的数据文件和统计文件是否存在
        File queueStatFile = new File(getQueueDataPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        return true;
    }

    // 把消息写入到队列对应的文件中，queue表示写入的队列，message表示写入的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        // 1. 检查要写入的队列对应的文件是否存在
        if (!checkFileExits(queue.getName())) {
            throw new MqException("MessageFileManager: 队列对应的文件不存在! queueName = " + queue.getName());
        }
        // 2. 把对象先序列化，转成二进制字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue) {
            // 3. 获取当前队列数据文件的文件长度，计算出改对象的offsetBeg和offsetEnd
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length()+4);
            message.setOffsetEnd(queueDataFile.length()+4+messageBinary.length);
            // 4. 写入消息到数据文件，这里是追加写
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                // 写入消息长度，占4个字节，使用位运算
//            outputStream.write((messageBinary.length >> 24) & 0xFF);
//            outputStream.write((messageBinary.length >> 16) & 0xFF);
//            outputStream.write((messageBinary.length >> 8) & 0xFF);
//            outputStream.write(messageBinary.length & 0xFF);
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    dataOutputStream.writeInt(messageBinary.length);
                    // 写入消息体
                    dataOutputStream.write(messageBinary);
                }
            }
            // 5，更新队列的统计信息
            Stat stat = readStat(queue.getName());
            stat.totalCount++;
            stat.validCount++;
            writeStat(queue.getName(), stat);
        }
    }

    // 删除消息，此时是逻辑删除，把消息的状态设置为无效
    public void deleteMessage(MSGQueue queue, Message message) throws IOException,ClassNotFoundException{
        synchronized (queue)  {
            // "rw":可读也可写
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                // 1.文件中读取对应的Message数据
                byte[] messageBinary = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(messageBinary);
                // 2. 把字节数组反序列化成Message对象
                Message diskMessage = (Message) BinaryTool.toObject(messageBinary);
                // 3. 修改Message对象的状态,isValid设置为无效
                // 此时不需要把message的isValid设置为无效，因为message是从内存中获取的,这个对象也要马上从内存中删除了
                diskMessage.setIsValid((byte) 0x0);
                // 4. 把修改后的Message对象重新序列化，写入到文件中
                byte[] newMessageBinary = BinaryTool.toBytes(diskMessage);
                // 虽然上面seek过了，但是上面seek完了，进行读操作，读操作会把光标往后移动
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(newMessageBinary);
                // 此时只有一个字节发生改变
            }
            // 5. 更新队列的统计信息
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) {
                stat.validCount--;
            }
            writeStat(queue.getName(), stat);
        }
    }

    // 读取所有的消息的内存，加载到内存中
    // 这个方法是为了初始化的时候，把消息加载到内存中
    // 这里的调用使其在程序启动时，不涉及多线程，所以不需要加锁，因此不需要锁对象，传入queueName即可
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                // 记录当前文件光标
                long currentOffset = 0;
                // 一个文件包含多个消息，此处需要循环读取
                while (true) {
                    // 1. 读取消息长度，这里可能读到文件末尾
                    // readInt()会抛出 EOFException 异常
                    int messageLength = dataInputStream.readInt();
                    // 2. 读取消息体
                    byte[] messageBinary = new byte[messageLength];
                    int actualLength =dataInputStream.read(messageBinary);
                    if (actualLength != messageLength) {
                        // 不匹配，说明文件有问题，格式错乱
                        throw new IOException("【MessageFileManager】 消息文件格式错误! queueName = " + queueName);
                    }
                    // 3. 反序列化，把字节数组转成Message对象
                    Message message = (Message) BinaryTool.toObject(messageBinary);
                    // 4. 判断消息是否有效
                    if (message.getIsValid() != 0x1) {
                        // 虽然是无效，但是光标还是需要移动的
                        currentOffset += (4+messageLength);
                        continue;
                    }
                    // 5. 有效数据，则需要把这个 Message 对象加入到链表中。加入之前还需要填写 offsetBeg 和 offsetEnd
                    //进行计算 offset 的时候，需要知道当前文件光标的位置的。由于当下使用的 DataInputStream 并不方便
                    //因此就需要手动计算下文件光标。
                    message.setOffsetBeg(currentOffset+4);
                    message.setOffsetEnd(currentOffset+4+messageLength);
                    // 移动光标
                    currentOffset += (4+messageLength);
                    messages.add(message);
                }
            }catch (EOFException e) {
                // 说明已经读到文件末尾了
                System.out.println("【MessageFileManager】 恢复Message成功! ");
            }
        }
        return messages;
    }

    // 检查当前队列消息数据是否需要进行垃圾回收
    public boolean checkGC(String queueName) {
        // 判断是否要进行垃圾回收，根据队列的统计信息来判断，即总消息数和有效消息数
        Stat stat = readStat(queueName);
        // 有效消息数小于总消息数的一半，并且总消息大于2000
        return stat.validCount < stat.totalCount / 2 && stat.totalCount > 2000;
    }

    // 新文件的位置
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "queue_data_new.txt";
    }

    // 进行垃圾回收，把无效的消息删除，把有效消息重新写入到新文件中
    // 记得把新文件重命名为旧文件名以及更新消息统计信息
    // 需要加锁，GC过程中，不允许有其他线程进行消息的写入和删除，否则会导致数据不一致,而且GC比较耗时
    public void gc(MSGQueue queue) throws IOException, ClassNotFoundException, MqException {
        synchronized (queue) {
            // 0. 由于GC比较耗时，统计下执行时间
            long startTime = System.currentTimeMillis();
            // 1. 创建新文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {
                // 如果新文件已经存在，说明上次GC没有执行完，程序异常退出了
                throw new MqException("【MessageFileManager】 上次GC没有执行完，程序异常退出了! queueName = " + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();
            if (!ok) {
                throw new MqException("【MessageFileManager】 创建新文件失败! queueDataNewName = " + queueDataNewFile.getAbsolutePath());
            }

            // 2. 读取旧文件，读取所有的有效消息对象,调用上述方法
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            // 3. 有效消息写入新文件，如果调用前面的sendMessage方法，需要使其支持文件名，以后可以改改，当作扩展
            try (OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            // 4. 删除+重命名新文件为旧文件名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if (!ok) {
                throw new MqException("【MessageFileManager】 删除旧文件失败! queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if (!ok) {
                throw new MqException("【MessageFileManager】 重命名新文件失败! queueDataNewFile = " + queueDataNewFile.getAbsolutePath()
                        + ",queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            // 5. 更新消息统计信息
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(), stat);

            long endTime = System.currentTimeMillis();
            System.out.println("【MessageFileManager】 GC执行完毕！queueName: " + queue.getName() + ",time=" + (endTime - startTime) + "ms");
        }
    }
}
